package org.codehaus.activemq.message;

import java.io.DataOutput;
import java.io.IOException;
import java.util.Hashtable;
import org.codehaus.activemq.util.BitArray;

public class ActiveMQMessageWriter extends AbstractPacketWriter
{
  public int getPacketType()
  {
    return 6;
  }

  public void writePacket(Packet packet, DataOutput dataOut)
    throws IOException
  {
    ActiveMQMessage msg = (ActiveMQMessage)packet;
    writeMessage(msg, dataOut);
  }

  protected void writeMessage(ActiveMQMessage msg, DataOutput dataOut)
    throws IOException
  {
    byte[] payload = msg.getBodyAsBytes();

    super.writeUTF(msg.getJMSMessageID(), dataOut);
    super.writeUTF(msg.getJMSClientID(), dataOut);
    super.writeUTF(msg.getProducerID(), dataOut);
    ActiveMQDestination.writeToStream((ActiveMQDestination)msg.getJMSDestination(), dataOut);
    dataOut.write(msg.getJMSDeliveryMode());
    dataOut.write(msg.getJMSPriority());

    BitArray ba = new BitArray();
    ba.set(0, msg.getJMSCorrelationID() != null);
    ba.set(1, msg.getJMSType() != null);
    ba.set(2, msg.getEntryBrokerName() != null);
    ba.set(3, msg.getEntryClusterName() != null);
    ba.set(4, msg.getTransactionId() != null);
    ba.set(5, msg.getJMSReplyTo() != null);
    ba.set(6, msg.getJMSTimestamp() > 0L);
    ba.set(7, msg.getJMSExpiration() > 0L);
    ba.set(8, msg.getJMSRedelivered());
    ba.set(9, msg.isXaTransacted());
    ba.set(10, msg.getConsumerNos() != null);
    ba.set(11, (msg.getProperties() != null) && (msg.getProperties().size() > 0));
    ba.set(12, payload != null);
    ba.writeToStream(dataOut);

    if (ba.get(0)) {
      super.writeUTF(msg.getJMSCorrelationID(), dataOut);
    }
    if (ba.get(1)) {
      super.writeUTF(msg.getJMSType(), dataOut);
    }
    if (ba.get(2)) {
      super.writeUTF(msg.getEntryBrokerName(), dataOut);
    }
    if (ba.get(3)) {
      super.writeUTF(msg.getEntryClusterName(), dataOut);
    }
    if (ba.get(4)) {
      super.writeUTF(msg.getTransactionId(), dataOut);
    }
    if (ba.get(5)) {
      ActiveMQDestination.writeToStream((ActiveMQDestination)msg.getJMSReplyTo(), dataOut);
    }
    if (ba.get(6)) {
      dataOut.writeLong(msg.getJMSTimestamp());
    }
    if (ba.get(7)) {
      dataOut.writeLong(msg.getJMSExpiration());
    }
    if (ba.get(10))
    {
      int[] cids = msg.getConsumerNos();
      dataOut.writeShort(cids.length);
      for (int i = 0; i < cids.length; i++) {
        dataOut.writeShort(cids[i]);
      }
    }
    if (ba.get(11)) {
      msg.writeMapProperties(msg.getProperties(), dataOut);
    }
    if (ba.get(12)) {
      dataOut.writeInt(payload.length);
      dataOut.write(payload);
    }
  }
}